package com.xhg.mqtt.core;

import com.xhg.mqtt.cache.CacheService;
import com.xhg.mqtt.constant.MqttConstants;
import com.xhg.mqtt.constant.MsgConstants;
import com.xhg.mqtt.dto.EventCodeEnum;
import com.xhg.mqtt.dto.MqttMessagePb;
import com.xhg.mqtt.job.MessageRepeatQueue;
import com.xhg.mqtt.job.ServicesPoolExecutor;
import com.xhg.mqtt.listener.MessageManager;
import com.xhg.mqtt.listener.MqttLoggerFactory;
import com.xhg.mqtt.util.ProtoBufUtil;
import java.text.MessageFormat;
import java.util.Arrays;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;

/**
 * 发布消息的回调类
 * <p>
 * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
 * 在回调中，将它用来标识已经启动了该回调的哪个实例。 必须在回调类中实现三个方法：
 * <p>
 * public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
 * <p>
 * public void connectionLost(Throwable cause)在断开连接时调用。
 * <p>
 * public void deliveryComplete(MqttDeliveryToken token)) 接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。 由
 * MqttClient.connect 激活此回调。
 */
public class Callback implements MqttCallbackExtended {

    /**
     * 连接断开回调
     */
    @Override
    public void connectionLost(Throwable cause) {
        MqttLoggerFactory.warn(MsgConstants.MQTT_BREAK_EXCEPTION, cause);
        try {
            MessageManager.reconnect(false);
            MessageRepeatQueue.getInstance().removeAllRecMsg();
            MessageRepeatQueue.getInstance().removeAllFailMsg();
        } catch (Exception e) {
            MqttLoggerFactory.error(MsgConstants.CONN_EXCEPTION, e);
        }
    }

    /**
     * 交付完成回调。在publish消息的时候会收到此回调. qos: 0 发送完则回调 1 或 2 会在对方收到时候回调
     *
     * @param token id,可通过id来获取发送的消息内容
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            token.setActionCallback(new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    System.out.println("成功了。。"+Arrays.toString(asyncActionToken.getTopics()));
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    System.out.println("失败了了。。"+asyncActionToken.getMessageId());
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("deliveryComplete---------" + token.isComplete());
        System.out.println("clientID---------" + token.getClient().getClientId());
        System.out.println("session---------" + token.getSessionPresent());
        System.out.println("messageId---------" + token.getMessageId());
    }

    /**
     * 消息接收回调，如果只发送，这里没关系
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) {
        try {
            if (!MqttClient.getMqttInstance().isConnected()) {
                return;
            }
            // subscribe后得到的消息会执行到这里面
            MqttLoggerFactory.info("mqtt sdk message:" + "接收消息主题 : " + topic);
            MqttLoggerFactory.info("mqtt sdk message:" + "接收消息Qos : " + message.getQos());
            //反序列化 具体业务处理
            MqttMessagePb.MqttMessage msg = ProtoBufUtil.decode(message.getPayload());

            //解析头信息
            MqttMessagePb.MqttHead head = msg.getHead();
            //指令码
            String eventCode = head.getEventCode();
            //设备id
            String deviceId = head.getDeviceId();
            String messageId = head.getMessageId();
            //判断topic是否是ack应答过来的
            if (topic.contains(MqttConstants.TOPC_ACK)) {
                MqttLoggerFactory.info(
                    "设备：" + deviceId + " 收到服务端ack回复 messageId：" + messageId + " 指令码:" + eventCode
                        + " 指令信息: " + EventCodeEnum.enumOf(eventCode).getDesc());
                MessageRepeatQueue.getInstance().remove(messageId);
            } else {
                if (CacheService.isTimeOut(messageId)) {
                    //解析body内容，这个要根据指令码才解析出具体是那个业务类
                    ServicesPoolExecutor.getThreadPools().execute(() -> {
                        Thread.currentThread().setName("消息处理线程");
                        MessageManager.messageArrived(message.getPayload());
                    });
                    CacheService.put(messageId,"1");
                }
            }
        } catch (Exception e) {
            MqttLoggerFactory.error(MsgConstants.CALLBACK_MESSAGE_EXCEPTION, e);
        }
    }

    /**
     * 链接成功后调用
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
//        try {
            //重连回调
            if (reconnect) {
                MqttLoggerFactory
                    .info(MessageFormat.format(MsgConstants.RECONNETC_SUCCESS, serverURI));
                //通知客户端重连成功
                MessageManager.reconnect(true);
                //重新订阅topic
                MqttClient.getInstance().subTopic();
            } else {
                //初始化连接首次回调
                MessageManager.connectComplete(MqttClient.getMqttInstance().isConnected());
            }
//        } catch (Exception e) {
//            MqttLoggerFactory.error(MsgConstants.CONN_EXCEPTION, e);
//        }
    }
}